package com.hzw.monitor.mysqlbinlog.handlers;

/**
 * 
 * @author zhiqiang.liu
 * @2016年1月14日
 * @qq:837500869
 */
import java.util.HashMap;
import java.util.List;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import com.hzw.monitor.mysqlbinlog.command.RequestBinaryLogCommand;
import com.hzw.monitor.mysqlbinlog.connection.ConnectionAttributes;
import com.hzw.monitor.mysqlbinlog.event.EventData;
import com.hzw.monitor.mysqlbinlog.event.EventHeader;
import com.hzw.monitor.mysqlbinlog.event.EventType;
import com.hzw.monitor.mysqlbinlog.event.data.RotateEventData;
import com.hzw.monitor.mysqlbinlog.netty.server.MyNioSocketChannel;
import com.hzw.monitor.mysqlbinlog.parser.DeleteRowsEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.EventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.FormatDescriptionParser;
import com.hzw.monitor.mysqlbinlog.parser.NullEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.QueryEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.RotateEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.RowsQueryEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.TableMapEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.UpdateRowsEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.WriteRowsEventDataParser;
import com.hzw.monitor.mysqlbinlog.parser.XidEventDataParser;
import com.hzw.monitor.mysqlbinlog.utils.ByteUtils;
import com.hzw.monitor.mysqlbinlog.utils.DataUtils;
import com.hzw.monitor.mysqlbinlog.utils.LoggerUtils;
import com.hzw.monitor.mysqlbinlog.utils.MyConstants;
import com.hzw.monitor.mysqlbinlog.utils.MyProperties;
import com.hzw.monitor.mysqlbinlog.utils.NettyUtils;
import com.hzw.monitor.mysqlbinlog.utils.TimeUtils;
import com.hzw.monitor.mysqlbinlog.zookeeper.ZooKeeperUtils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class BinlogEventParseHandler extends SimpleChannelInboundHandler<ByteBuf> {
	// http://dev.mysql.com/doc/internals/en/binlog-event.html
	private static final Logger logger = LogManager.getLogger(BinlogEventParseHandler.class);
	private static final EventType[] EVENT_TYPES = EventType.values();
	private boolean enterLargeBlockPacket = false;// 暂时不支持这种情况,TODO
	private HashMap<EventType, EventDataParser> parsers = new HashMap<EventType, EventDataParser>();

	public BinlogEventParseHandler() {
		// null_event
		parsers.put(EventType.UNKNOWN, new NullEventDataParser());
		// rotate
		parsers.put(EventType.ROTATE, new RotateEventDataParser());
		// format_description
		parsers.put(EventType.FORMAT_DESCRIPTION, new FormatDescriptionParser());
		// query
		parsers.put(EventType.QUERY, new QueryEventDataParser());
		// table_map
		parsers.put(EventType.TABLE_MAP, new TableMapEventDataParser());
		// delete_rows
		parsers.put(EventType.DELETE_ROWS, new DeleteRowsEventDataParser(false));
		// ext_delete_rows
		parsers.put(EventType.EXT_DELETE_ROWS, new DeleteRowsEventDataParser(true));
		// write_rows
		parsers.put(EventType.WRITE_ROWS, new WriteRowsEventDataParser(false));
		// ext_write_rows
		parsers.put(EventType.EXT_WRITE_ROWS, new WriteRowsEventDataParser(true));
		// update_rows
		parsers.put(EventType.UPDATE_ROWS, new UpdateRowsEventDataParser(false));
		// ext_update_rows
		parsers.put(EventType.EXT_UPDATE_ROWS, new UpdateRowsEventDataParser(true));
		// xid
		parsers.put(EventType.XID, new XidEventDataParser());
		// rows_query
		parsers.put(EventType.ROWS_QUERY, new RowsQueryEventDataParser());
	}

	private void handleRotate(EventType eventType, EventHeader header, EventData eventData,
			ConnectionAttributes myAttributes, ChannelHandlerContext context) {
		// newValue
		RotateEventData rotateData = (RotateEventData) eventData;
		String newBinlogFileName = rotateData.getBinlogFilename();
		long newBinlogPosition = rotateData.getBinlogPosition();
		String newValue = newBinlogFileName + ":" + newBinlogPosition;
		// localValue
		String localValue = myAttributes.getBinlogFileName() + ":" + myAttributes.getBinlogPosition();
		// 条件更新
		if (false == newValue.equals(localValue)) {
			myAttributes.updateBinlogNameAndPosition(newBinlogFileName, newBinlogPosition);
			ZooKeeperUtils.update(myAttributes.getBinlogPositionZKPath(), newValue + ":" + System.currentTimeMillis());
			myAttributes.setAccumalatedBinlogPositionCount(0);
			// 发送命令
			new RequestBinaryLogCommand(myAttributes.getClientId(), newBinlogFileName, newBinlogPosition)
					.write(context);
			LoggerUtils.debug(logger, "receive rotate event,[BinlogEventParseHandler] send RequestBinaryLogCommand: "
					+ newBinlogFileName + ":" + newBinlogPosition);
		} else {
			// 直接忽略,主要是启动时有一个默认的rotate事件
			LoggerUtils.debug(logger, "recive rotate event,the same as local data,omit");
		}

	}

	private void handleIO(EventType eventType, EventHeader header, EventData eventData,
			ConnectionAttributes myAttributes, ChannelHandlerContext context) throws Exception {
		LoggerUtils.debug(logger, eventType.toString());
		// 0)先处理数据,处理完了才进行位置记录
		boolean result = DataUtils.handle(myAttributes, header, eventData);
		// 保存出错，//更新到ZK,立刻返回
		if (false == result) {
			// 如果是后端存储错误，则不要频繁试错
			TimeUtils.sleepSeconds(1);
			throw new Exception("save data error, throw new Exception...");
		}
		// 1)存储成功了就更新内存变量
		myAttributes.updateBinlogNameAndPosition(myAttributes.getBinlogFileName(), header.getNextPosition());
		// 2)按需更新ZK
		if (header.getNextPosition() > 0) {
			if (myAttributes.incrAccumalatedBinlogPositionCount() >= MyProperties.getInstance()
					.getAccumalatedCountValue()) {// 超过阀值
				// 更新到ZK
				ZooKeeperUtils.update(myAttributes.getBinlogPositionZKPath(), myAttributes.getBinlogFileName() + ":"
						+ myAttributes.getBinlogPosition() + ":" + System.currentTimeMillis());
				// 务必重置计数器
				myAttributes.setAccumalatedBinlogPositionCount(0);
				LoggerUtils.debug(logger, "update binlogPath[zk]---" + myAttributes.getBinlogFileName() + ":"
						+ myAttributes.getBinlogPosition());
			}
		} else {
			LoggerUtils.error(logger, "nextBinlogPosition:" + header.getNextPosition());
		}
	}

	private void handlerDefault(EventType eventType, EventHeader header, EventData eventData,
			ConnectionAttributes myAttributes, ChannelHandlerContext context) {
		// 其它不处理的事情都在这里
		// 这里其实不能写业务[比如更新本地内存位置]，否则下次获取会出错
	}

	private void handle(EventType eventType, EventHeader header, EventData eventData, ConnectionAttributes myAttributes,
			ChannelHandlerContext context) throws Exception {
		// 保证参数OK
		if (null == eventType || null == eventData || null == myAttributes) {
			return;
		}
		switch (eventType) {
		case ROTATE:
			handleRotate(eventType, header, eventData, myAttributes, context);
			break;
		case WRITE_ROWS:
		case EXT_WRITE_ROWS:
		case UPDATE_ROWS:
		case EXT_UPDATE_ROWS:
		case DELETE_ROWS:
		case EXT_DELETE_ROWS:
			handleIO(eventType, header, eventData, myAttributes, context);
			break;
		default:// 其它事件如何处理
			handlerDefault(eventType, header, eventData, myAttributes, context);
			break;
		}

	}

	private void handleZKHistoryRecord(EventHeader header, ConnectionAttributes myAttributes) {
		// 每一条都会处理,不区分具体事件类型
		try {
			long t = header.getTimestamp();
			String day = TimeUtils.getTimeDay(t);

			myAttributes.setCurrentPosition(myAttributes.getNextPosition());
			myAttributes.setNextPosition(header.getNextPosition());
			// 第一次进来先记下当前天
			if (myAttributes.getHistoryPositionDay() == null) {
				myAttributes.setHistoryPositionDay(day);
			} else if (!day.equals(myAttributes.getHistoryPositionDay())) {
				// 需要记录到zk
				String historyPath = myAttributes.getBinlogPositionZKPath() + "/" + day;
				if (!ZooKeeperUtils.exist(historyPath)) {
					ZooKeeperUtils.create(historyPath, myAttributes.getBinlogFileName() + ":"
							+ myAttributes.getCurrentPosition() + ":" + System.currentTimeMillis());
				} else {
					ZooKeeperUtils.update(historyPath, myAttributes.getBinlogFileName() + ":"
							+ myAttributes.getCurrentPosition() + ":" + System.currentTimeMillis());
				}
				myAttributes.setHistoryPositionDay(day);
				// 并判断zk是否保存了30天的数据，如果超过则删除
				List<String> children = ZooKeeperUtils.getChildren(myAttributes.getBinlogPositionZKPath());
				for (String child : children) {
					if (((t - TimeUtils.getDayTime(child))
							/ MyConstants.MILLISECOND_OF_ONE_DAY) > MyConstants.HISTORY_RECORD_DAY) {
						ZooKeeperUtils.deletePath(myAttributes.getBinlogPositionZKPath() + "/" + child);
					}
				}
			}
		} catch (Exception e) {
			LoggerUtils.error(logger, e.toString());
		}
	}

	@Override
	protected void channelRead0(ChannelHandlerContext context, ByteBuf msg) throws Exception {
		//
		try {
			// LoggerUtils.debug(logger, "进入BinLogEventParseHandler处理器");
			EventHeader header = new EventHeader();
			header.setTimestamp(ByteUtils.readUnsignedLong(msg, 4) * 1000L);
			header.setEventType(EVENT_TYPES[ByteUtils.readUnsignedInt(msg, 1)]);
			header.setServerId(ByteUtils.readUnsignedLong(msg, 4));
			header.setEventLength(ByteUtils.readUnsignedLong(msg, 4));
			header.setNextPosition(ByteUtils.readUnsignedLong(msg, 4));
			header.setFlag(ByteUtils.readUnsignedInt(msg, 2));
			// LoggerUtils.debug(logger, header.toString());
			// 2)获取EventParser
			EventDataParser parser = parsers.get(header.getEventType());
			ConnectionAttributes myAttributes = ((MyNioSocketChannel) context.channel()).getAttributes();
			int checksumLength = (int) myAttributes.getChecksumLength();
			EventData eventData = parser.parse(msg, context, checksumLength);
			// LoggerUtils.debug(logger, eventData.toString());
			// 3)关联
			eventData.setEventHeader(header);
			// 4)处理
			// 决定是否更新binlog[name:position]的历史记录
			handleZKHistoryRecord(header, myAttributes);
			// 5)处理各种事件
			handle(header.getEventType(), header, eventData, myAttributes, context);

		} catch (Exception e) {
			LoggerUtils.error(logger, e.toString());
			throw new Exception(e);
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		// Close the connection when an exception is raised.
		// cause.printStackTrace();//务必要关闭
		LoggerUtils.error(logger, cause.toString());
		NettyUtils.cleanChannelContext(ctx, cause);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		LoggerUtils.debug(logger, "[channelInactive] socket is closed by remote server");
		NettyUtils.cleanChannelContext(ctx, null);
	}

}
